package com.shujia.redis

import org.apache.spark.sql.DataFrame
import redis.clients.jedis.Jedis
import com.shujia.tour.Constants

object IndexToRedis {
  /**
    * 将指标写入redis
    *
    * @param df    数据
    * @param c1    区域字段
    * @param c2    指标指定
    * @param dayId 时间
    */

  def fit(df: DataFrame, c1: String, c2: String, dayId: String): Unit = {

    /**
      * 性别按天 [省id,性别,客流量]
      *
      */
    val groupDF = df.groupBy(c1, c2).count()

    groupDF.rdd.map(row => {
      val pId = row.getAs[String](c1)
      val cname = row.getAs[String](c2)
      val flow = row.getAs[Long]("count")
      (pId, cname + ":" + flow)

    })
      //(850,2:152|1:135)
      .reduceByKey(_ + "|" + _)
      .foreachPartition(iter => {
        //创建redis连接
        val jedis = new Jedis(Constants.REDIS_HOST, Constants.REDIS_PORT)
        jedis.select(10)

        iter.foreach {
          case (pid: String, cname: String) => {
            val key = pid + "_" + dayId
            //将性别客流量写入redis
            jedis.hset(key, c2 + "_flow", cname)
          }
        }
        jedis.close()
      })
  }
}
